package com.shujia.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.storage.StorageLevel

object SparkOnHiveDemo2 {
  def main(args: Array[String]): Unit = {
    val ss: SparkSession = SparkSession.builder()
      .config("spark.sql.shuffle.partitions", "1") // 参数设置优先级：代码 > 命令参数 > 环境配置
      .master("local") // 提交到服务器中执行的话，不需要指定local
      .appName("spark操作hive数据")
      .enableHiveSupport() //读取配置文件，连接hive
      .getOrCreate()

//    ss.sql("use bigdata33")
//    ss.sql("select clazz,count(1) as counts from students group by clazz").show()

    val studentDF: DataFrame = ss.read
      .format("csv")
      .option("sep", ",")
      .schema("id STRING,name STRING,age INT,gender STRING,clazz STRING")
      .load("spark/data/students.txt")
//    studentDF.createOrReplaceTempView("students2")


//    val i = 10
    //spark sql每次sql执行的时候，相当于之前core中一个DAG有向无环图，重复计算第一个DF。
    //如果与hive结合的话，每次执行sql语句都需要从hive中加载数据【实际上就是从hdfs某个目录加载（磁盘）】
    //可以单独对重复使用DF做缓存，这样每次行sql语句的时候，从spark本地缓存中拿数据，加快执行速度
//    studentDF.persist(StorageLevel.MEMORY_AND_DISK_SER_2)
    val scoreDF: DataFrame = ss.read
      .format("csv")
      .option("sep", ",")
      .schema("id STRING,subject_id STRING,score INT")
      .load("spark/data/score.txt")
//    scoreDF.createOrReplaceTempView("scores2")

//    ss.sql(
//      """
//        |select
//        |*
//        |from
//        |students2 t1
//        |join
//        |scores2 t2
//        |on (t1.id=t2.id)
//        |
//        |""".stripMargin).show()

    import ss.implicits._
    import org.apache.spark.sql.functions._

    //字符串拼接
    //    studentDF.select($"id",$"name",concat(expr("'数加: '"),$"name"))
    //      .show()

    //开窗
    scoreDF.groupBy("id")
      .agg( //agg可以在分组后同时使用多个聚合函数
        sum("score") as "sumScore",
        count(expr("1")) as "num"
      ).join(studentDF,"id")
      //如果要新增一列 使用withColumn
      .withColumn("rn", row_number() over Window.partitionBy("clazz").orderBy($"sumScore".desc))
      .withColumn("tmp_col",expr("'方直'"))
      .show()



  }
}
